/*
package com.rinlink.gateway.jt808.jt808;

import com.alibaba.fastjson.JSONObject;
import com.rinlink.gateway.jt808.model.TerminalRequestMessage;
import com.rinlink.gateway.jt808.model.Jt808Header;
import io.github.hylexus.jt808.dispatcher.RequestMsgDispatcher;
import io.github.hylexus.jt808.msg.RequestMsgMetadata;
import io.github.hylexus.jt808.msg.RequestMsgWrapper;
import org.springframework.beans.BeanUtils;
import org.springframework.kafka.core.KafkaTemplate;

import java.util.Base64;

public class MessageToKafka implements RequestMsgDispatcher {

    private KafkaTemplate kafkaTemplate;

    private String topic;

    public MessageToKafka(KafkaTemplate kafkaTemplate,String topic) {
        this.kafkaTemplate = kafkaTemplate;
        this.topic = topic;
    }

    @Override
    public void doDispatch(RequestMsgWrapper wrapper) throws Exception {

        RequestMsgMetadata msgMetadata = wrapper.getMetadata();

        String body = Base64.getEncoder().encodeToString(msgMetadata.getBodyBytes());

        Jt808Header header = new Jt808Header();
        BeanUtils.copyProperties(msgMetadata.getHeader(),header);

        TerminalRequestMessage msg = new TerminalRequestMessage()
            .setBody(body)
            .setMsgId(wrapper.getMetadata().getMsgType().getMsgId())
            .setHeader(header);
        kafkaTemplate.send(topic,JSONObject.toJSONString(msg));
    }

}
*/
